package com.we.flink.userlabelplatform.tag;

import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.util.Collector;

import com.alibaba.fastjson.JSONObject;
import com.we.flink.userlabelplatform.table.TableWechatHaofenqiUserInfo;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class TagWechatHaofenqiUserInfo {
    public long uid;
    /** null：未绑定 0：绑定但未关注公众号 1：绑定且关注公众号 */
    public int wechat_hfq_user_status;

    public int updated_at;

    public TagWechatHaofenqiUserInfo(long uid, int wechat_hfq_user_status, int updated_at) {
        this.uid = uid;
        this.wechat_hfq_user_status = wechat_hfq_user_status;
        this.updated_at = updated_at;
    }

    public static TagWechatHaofenqiUserInfo init(long uid) {
        return new TagWechatHaofenqiUserInfo(uid, -1, 0);
    }

    public static Logger LOG = LoggerFactory.getLogger(TagTieCardRiskTag.class);

    public static SingleOutputStreamOperator<String> getFlatMapStream(
            StreamTableEnvironment tableEnv, StreamExecutionEnvironment env) {
        Table tableRes = tableEnv.sqlQuery(TableWechatHaofenqiUserInfo.QUERYSQL);
        DataStream<Tuple2<Boolean, TableWechatHaofenqiUserInfo>> inputStream =
                tableEnv.toRetractStream(tableRes, TableWechatHaofenqiUserInfo.class);

        SingleOutputStreamOperator<String> out =
                inputStream
                        .filter(
                                new FilterFunction<Tuple2<Boolean, TableWechatHaofenqiUserInfo>>() {
                                    @Override
                                    public boolean filter(
                                            Tuple2<Boolean, TableWechatHaofenqiUserInfo> value)
                                            throws Exception {
                                        return value.f0;
                                    }
                                })
                        .uid("TagWechatHaofenqiUserInfo-filter")
                        .keyBy(
                                new KeySelector<
                                        Tuple2<Boolean, TableWechatHaofenqiUserInfo>, Long>() {
                                    @Override
                                    public Long getKey(
                                            Tuple2<Boolean, TableWechatHaofenqiUserInfo> value)
                                            throws Exception {
                                        return value.f1.getUid();
                                    }
                                })
                        .flatMap(
                                new RichFlatMapFunction<
                                        Tuple2<Boolean, TableWechatHaofenqiUserInfo>, String>() {
                                    ValueState<TagWechatHaofenqiUserInfo> tagWechatHfqUserInfoState;

                                    @Override
                                    public void open(Configuration parameters) throws Exception {
                                        tagWechatHfqUserInfoState =
                                                getRuntimeContext()
                                                        .getState(
                                                                new ValueStateDescriptor<>(
                                                                        "TagWechatHaofenqiUserInfo",
                                                                        TagWechatHaofenqiUserInfo
                                                                                .class));
                                    }

                                    @Override
                                    public void flatMap(
                                            Tuple2<Boolean, TableWechatHaofenqiUserInfo> value,
                                            Collector<String> out)
                                            throws Exception {
                                        long uid = value.f1.getUid();
                                        /** 和业务表枚举值相反 0:未关注 1：关注 */
                                        int status = (value.f1.getStatus() + 1) % 2;
                                        int updated_at = value.f1.getUpdated_at();

                                        TagWechatHaofenqiUserInfo tmpSatate =
                                                tagWechatHfqUserInfoState.value();
                                        TagWechatHaofenqiUserInfo curState =
                                                tmpSatate == null
                                                        ? TagWechatHaofenqiUserInfo.init(uid)
                                                        : tmpSatate;

                                        if (curState.getUpdated_at() < updated_at) {
                                            if (curState.getWechat_hfq_user_status() != status) {
                                                curState.wechat_hfq_user_status = status;
                                                curState.updated_at = updated_at;
                                                JSONObject jsonObject = new JSONObject();
                                                jsonObject.put("uid", uid);
                                                jsonObject.put("wechat_hfq_user_status", status);

                                                out.collect(jsonObject.toString());
                                                tagWechatHfqUserInfoState.update(curState);
                                            }
                                        }
                                    }
                                });

        return out;
    }

    public int getUpdated_at() {
        return updated_at;
    }

    public int getWechat_hfq_user_status() {
        return wechat_hfq_user_status;
    }
}
